package cn.liwq.study.demo.netty.lab.mahjong.server.core;

import cn.liwq.study.demo.netty.lab.mahjong.common.domain.Player;
import cn.liwq.study.demo.netty.lab.mahjong.common.dto.CreateRoomRequest;
import cn.liwq.study.demo.netty.lab.mahjong.common.dto.EnterRoomRequest;
import cn.liwq.study.demo.netty.lab.mahjong.common.enums.MahjongProcessorManager;
import cn.liwq.study.demo.netty.lab.mahjong.common.protocol.MahjongMessage;
import cn.liwq.study.demo.netty.lab.mahjong.common.protocol.MahjongProtocol;
import cn.liwq.study.demo.netty.lab.mahjong.common.protocol.MahjongProtocolHeader;
import cn.liwq.study.demo.netty.lab.mahjong.common.utils.IdUtils;
import cn.liwq.study.demo.netty.lab.mahjong.server.context.MahjongContext;
import cn.liwq.study.demo.netty.lab.mahjong.server.data.DataManager;
import cn.liwq.study.demo.netty.lab.mahjong.server.processor.MahjongProcessor;
import io.netty.channel.Channel;
import io.netty.util.NettyRuntime;
import io.netty.util.concurrent.DefaultEventExecutor;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.EventExecutorChooserFactory;
import io.netty.util.concurrent.MultithreadEventExecutorGroup;

import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author lweiqiang
 * @date 2021/5/10
 */
public class MahjongEventExecutorGroup extends MultithreadEventExecutorGroup {

    private MahjongEventExecutorGroup(int nThreads) {
        // 使用自定义的选择器工厂
        super(nThreads, null, new MahjongEventExecutorChooserFactory(), null);
    }

    @Override
    protected EventExecutor newChild(Executor executor, Object... args) throws Exception {
        return new DefaultEventExecutor(this, executor);
    }

    // 实例，单例模式
    private static final MahjongEventExecutorGroup INSTANCE =
            new MahjongEventExecutorGroup(NettyRuntime.availableProcessors());

    // 静态方法，MahjongServerHandler中直接调用该方法
    public static void execute(Channel channel, MahjongProtocol mahjongProtocol) {
        // 协议头
        MahjongProtocolHeader header = mahjongProtocol.getHeader();
        // 协议体
        MahjongMessage message = (MahjongMessage) mahjongProtocol.getBody();
        // 创建房间和加入房间需要做一些处理
        Long roomId;
        if (message instanceof CreateRoomRequest) {
            // 如果是创建房间消息，生成一个roomId，并将当前channel与之绑定
            roomId = IdUtils.generateId();
            // 一个channel一旦建立，始终与一个eventLoop绑定
            // 所以，可以把channel与房间号的绑定放在线程本地缓存中
            MahjongContext.currentContext().setChannelRoomId(channel, roomId);
        } else if (message instanceof EnterRoomRequest) {
            // 如果是加入房间消息，将当前channel与传入的tableId绑定
            EnterRoomRequest enterRoomRequest = (EnterRoomRequest) message;
            roomId = enterRoomRequest.getRoomId();
            MahjongContext.currentContext().setChannelRoomId(channel, roomId);
        } else {
            // 其它消息则从context中获取当前channel对应的房间号
            // 当然，也可能没有，比如登录请求
            roomId = MahjongContext.currentContext().getChannelRoomId(channel);
        }

        // 设置房间id到context中，以便next()方法可以取到
        MahjongContext.currentContext().setCurrentRoomId(roomId);
        // 将消息扔到业务线程池中处理
        INSTANCE.execute(
                () -> {
                    // 已经切换线程，重新设置房间号到context中
                    MahjongContext.currentContext().setCurrentRoomId(roomId);
                    // 设置channel等其它线程本地变量
                    MahjongContext.currentContext().setCurrentChannel(channel);
                    MahjongContext.currentContext().setChannelRoomId(channel, roomId);
                    MahjongContext.currentContext().setRequestHeader(header);
                    MahjongContext.currentContext()
                            .setCurrentRoom(MahjongContext.currentContext().getRoomById(roomId));

                    Player currentPlayer = DataManager.getChannelPlayer(channel);
                    if (currentPlayer != null) {
                        MahjongContext.currentContext().setCurrentPlayer(currentPlayer);
                        MahjongContext.currentContext().setPlayerChannel(currentPlayer, channel);
                    }
                    // 寻找处理器
                    MahjongProcessor processor = MahjongProcessorManager.choose(message);
                    if (processor != null) {
                        // 交给处理器处理
                        processor.process(message);
                    } else {
                        throw new RuntimeException(
                                "not found processor, msgType=" + message.getClass().getName());
                    }
                });
    }

    /** 工厂类 */
    private static class MahjongEventExecutorChooserFactory implements EventExecutorChooserFactory {

        @Override
        public EventExecutorChooser newChooser(EventExecutor[] executors) {
            return new MahjongEventExecutorChooser(executors);
        }
        // 真正使用的选择器
        private static class MahjongEventExecutorChooser
                implements EventExecutorChooserFactory.EventExecutorChooser {

            private final AtomicInteger idx = new AtomicInteger();
            private final EventExecutor[] executors;

            MahjongEventExecutorChooser(EventExecutor[] executors) {
                this.executors = executors;
            }

            @Override
            public EventExecutor next() {
                // 从上下文中取出当前的房间id
                Long roomId = MahjongContext.currentContext().getCurrentRoomId();
                long id;
                if (roomId != null) {
                    id = roomId;
                } else {
                    // 没获取到房间号的消息轮徇扔到业务线程池中处理
                    // 他们往往跟房间信息没啥关系，比如登录请求
                    id = idx.getAndIncrement();
                }
                // 根据id取模选择线程执行
                return executors[(int) (id & executors.length - 1)];
            }
        }
    }
}
